package com.chengv.redismq.core;

import cn.hutool.system.SystemUtil;
import com.chengv.redismq.annotation.RedisMQMessageListener;
import com.chengv.redismq.util.MethodUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.lang.NonNull;

import java.lang.reflect.Method;
import java.time.Duration;

/**
 * 消息监听器注册器
 *
 * @author chengv
 * @see RedisMQMessageListenerAnnotationPostProcessor
 */
@Slf4j
public class RedisMQStreamListenerRegistrar implements ApplicationContextAware, InitializingBean {

    private ConfigurableApplicationContext applicationContext;
    private final RedisTemplate<String, String> redisTemplate;
    private final RedisSerializer<Object> hashValueSerializer;
    private final ConfigurableEnvironment environment;
    private StreamMessageListenerContainer<String, ObjectRecord<String, String>> container;

    public RedisMQStreamListenerRegistrar(RedisTemplate<String, String> redisTemplate,
                                          RedisSerializer<Object> hashValueSerializer, ConfigurableEnvironment environment) {
        this.redisTemplate = redisTemplate;
        this.hashValueSerializer = hashValueSerializer;
        this.environment = environment;

    }

    @Override
    public void setApplicationContext(@NonNull ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = (ConfigurableApplicationContext) applicationContext;
    }

    @Override
    public void afterPropertiesSet() {
        registerContainer(redisTemplate);
    }

    private void registerContainer(RedisTemplate<String, String> redisTemplate) {
        // 创建容器
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder()
                        .batchSize(10) // 单次拉取消息数量
                        .pollTimeout(Duration.ofSeconds(2)) // 拉取消息的超时时间
                        .targetType(String.class) // 拉取的消息的类型。使用 JSON 字符串，生产者、消费者自行转换，两者需要一致
                        .build();
        StreamMessageListenerContainer<String, ObjectRecord<String, String>> container =
                StreamMessageListenerContainer.create(redisTemplate.getRequiredConnectionFactory(), options);

        // 注册容器 Bean
        GenericApplicationContext genericApplicationContext = (GenericApplicationContext) applicationContext;
        genericApplicationContext.registerBean(StreamMessageListenerContainer.class, () -> container, (beanDefinition) -> {
            beanDefinition.setInitMethodName("start");
            beanDefinition.setDestroyMethodName("stop");
        });
        //noinspection unchecked,UnnecessaryLocalVariable
        StreamMessageListenerContainer<String, ObjectRecord<String, String>> containerBean =
                genericApplicationContext.getBean(StreamMessageListenerContainer.class);
        this.container = containerBean;
    }

    /**
     * 注册 {@link RedisMQStreamListener} 到 {@link StreamMessageListenerContainer}
     *
     * @param bean       {@link RedisMQMessageListener} 标注的类
     * @param method     {@link RedisMQMessageListener} 标注的方法
     * @param annotation {@link RedisMQMessageListener}
     */
    public void register(Object bean, Method method, RedisMQMessageListener annotation) {
        // 创建 RedisMQStreamListener
        RedisMQStreamListener listener = createRedisMQStreamListener(bean, method, annotation);
        String streamKey = listener.getStreamKey();
        String consumerGroup = listener.getConsumerGroup();

        // 构建消费者名称
        String consumerName = buildConsumerName();

        // 如果该主题下不存在此消费者组，则创建一个
        createGroupIfNotExist(redisTemplate, streamKey, consumerGroup);

        // 注册 RedisMQStreamListener 到 StreamMessageListenerContainer
        StreamMessageListenerContainer.ConsumerStreamReadRequest<String> streamRequest =
                StreamMessageListenerContainer.StreamReadRequest
                        // 消费进度，设置为从上次最后一个消费的开始
                        .builder(StreamOffset.create(streamKey, ReadOffset.lastConsumed()))
                        .consumer(Consumer.from(consumerGroup, consumerName)) // 消费者
                        .autoAcknowledge(false) // 不自动 ack 确认消费
                        // 默认配置为 true，发生异常就取消消费，显然不符合预期；因此，我们设置为 false
                        .cancelOnError(throwable -> false)
                        .build();
        container.register(streamRequest, listener);
        log.info("[registerRedisMQStreamListener][成功], listener=({}), streamKey=({}), " +
                "consumerGroup=({}), consumerName=({})", MethodUtils.getFullName(listener.getListenerBean(),
                listener.getListenerMethod()), streamKey, consumerGroup, consumerName);
    }

    private RedisMQStreamListener createRedisMQStreamListener(Object bean, Method method, RedisMQMessageListener annotation) {
        return new RedisMQStreamListener(bean, method, redisTemplate, hashValueSerializer,
                environment.resolvePlaceholders(annotation.topic()), environment.resolvePlaceholders(annotation.consumerGroup()));
    }

    /**
     * 如果该主题下不存在此消费者组，则创建一个
     *
     * @param redisTemplate Redis 模板
     * @param streamKey     streamKey
     * @param consumerGroup 消费者组
     */
    private void createGroupIfNotExist(RedisTemplate<String, String> redisTemplate, String streamKey, String consumerGroup) {
        if (Boolean.FALSE.equals(redisTemplate.hasKey(streamKey))) {
            redisTemplate.opsForStream().createGroup(streamKey, consumerGroup);
        } else {
            if (redisTemplate.opsForStream().groups(streamKey).stream().noneMatch(e -> e.groupName().equals(consumerGroup))) {
                redisTemplate.opsForStream().createGroup(streamKey, consumerGroup);
            }
        }
    }

    /**
     * 构建消费者名称，使用本地 IP + 进程编号的方式
     */
    private String buildConsumerName() {
        return String.format("%s@%d", SystemUtil.getHostInfo().getAddress(), SystemUtil.getCurrentPID());
    }
}
